package com.headcaselabs.lb.director;

import java.io.Serializable;
import java.rmi.RemoteException;
import java.rmi.registry.LocateRegistry;
import java.rmi.registry.Registry;
import java.rmi.server.UnicastRemoteObject;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.LinkedList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ThreadPoolExecutor;

import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.CommandLineParser;
import org.apache.commons.cli.GnuParser;
import org.apache.commons.cli.HelpFormatter;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.OptionBuilder;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.ParseException;
import com.headcaselabs.lb.common.Director;
import com.headcaselabs.lb.common.LBQuery;
import com.headcaselabs.lb.common.PipeCallback;
import com.headcaselabs.lb.common.Record;
import com.headcaselabs.lb.common.Server;

/* 
Copyright 2007, HeadCase Humanufacturing, Inc.

	Licensed under the Apache License, Version 2.0 (the "License"); you
	may not use this file except in compliance with the License. You may
	obtain a copy of the License at
	
	   http://www.apache.org/licenses/LICENSE-2.0
	
	Unless required by applicable law or agreed to in writing, software
	distributed under the License is distributed on an "AS IS" BASIS,
	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
	implied. See the License for the specific language governing
	permissions and limitations under the License.
*/
/**
 * The director responsible for Load Balancing. Implements the publisher /
 * subscriber pattern, so that clients who would like to receive notificatioin
 * about servers, can sign up. The Director tries to create a registry and binds
 * itself to this registry.
 * 
 * @author Florian Leibert
 * 
 */
public class DirectorImpl extends UnicastRemoteObject implements Director,
		Serializable {
	private static final long											serialVersionUID	= 7867817766044955932L;

	// Check every 15 minutes for updates
	protected static long													checkInterval			= 900000;

	private long																	lastModify				= 0;
	

	// private List<ServerSetIf> serverSets;
	protected ConcurrentHashMap<String, List<Server>>			servers;
	protected ConcurrentHashMap<String, ? extends LoadBalancedRecord> lbtable;
	protected static ConcurrentHashMap<LBQuery, Record> mappings;
	protected Object															lock;
	protected HashSet<Record> activeRecords;
	// protected List<PipeCallback> subscribers;
	protected ConcurrentLinkedQueue<PipeCallback>	subscribers;
	
	protected ThreadPoolExecutor threadPool;
	
	protected DBTools															dbtools;

	// Suboptimal Director <-> DBPoll are too cohesive.
	// private Thread dbPollThread;

	protected static String												dbHost						= "localhost";

	protected static String												dbUser						= "powerpipe";

	protected static String												dbPwd							= "powerpipe";

	protected static String												db								= "powerpipe";

	protected final static String									SERVER_TABLE			= "servers";
	
	public static final int POOL_SIZE = 20;
	
	
	Random random = new Random();
	
	/**
	 * Constructor
	 * @throws RemoteException
	 */
	public DirectorImpl(DBTools dbtools) throws RemoteException {
		super();
		this.servers = new ConcurrentHashMap<String, List<Server>>();
		// subscribers = new LinkedList<PipeCallback>();
		subscribers = new ConcurrentLinkedQueue<PipeCallback>();
		this.dbtools = dbtools;
		lock = new Object();
		mappings = new ConcurrentHashMap<LBQuery, Record>();
		lbtable = new ConcurrentHashMap<String, LoadBalancedRecord>();
		activeRecords = new HashSet<Record>();
		random = new Random();
		DBPoll dbpoll = new DBPoll(this, dbtools);
		Thread dbPollThread = new Thread(dbpoll);
		dbPollThread.start();
	}
	

	/*
	 * (non-Javadoc)
	 * @see com.headcaselabs.lb.common.Director#checkForUpdates()
	 */
	public void checkForUpdates() throws RemoteException {
		synchronized (lock) {
			log("Notifying DBPolling Process");
			lock.notify();
		}
	}

	/*
	 * (non-Javadoc)
	 * @see com.headcaselabs.lb.common.Director#subscribe(com.headcaselabs.lb.common.PipeCallback)
	 */
	public synchronized void subscribe(PipeCallback client) throws RemoteException {
		log("New Subscriber:" + client);
		subscribers.add(client);
		//for (String groupName : servers.keySet()) {
		//	client.updateServers(groupName, servers.get(groupName));
		//}
	}

	/*
	 * (non-Javadoc)
	 * @see com.headcaselabs.lb.common.Director#unsubscribe(com.headcaselabs.lb.common.PipeCallback)
	 */
	public synchronized void unsubscribe(PipeCallback client)
			throws RemoteException {
		log("Unsubscribing client:" + client);
		subscribers.remove(client);
	}

	
	/**
	 * Terminates all clients ungracefully
	 */
	public void killClients()	{
		for (PipeCallback client : subscribers) {
				try {
					client.quit();
				} catch (Throwable t) {
					log("Removing client:" + client);
					subscribers.remove(client);
				}
		}
	}
	
	
	//TODO: implement group based record updates
	public synchronized void updateRecords() {
		RandomizedRoundRobin rs;
		ConcurrentHashMap<String, LoadBalancedRecord> lbtable = new ConcurrentHashMap<String, LoadBalancedRecord>();
		activeRecords = new HashSet<Record>();
		for (String groupName : servers.keySet()) {
			//Currently, only RandomizedRoundRobin is implemented
			rs = new RandomizedRoundRobin(groupName, random);
			rs.buildEntries(servers.get(groupName));
			lbtable.put(groupName, rs);
			activeRecords.addAll(rs.getRecords());
		}
		
		for (LBQuery lbq: mappings.keySet()) {
			if (!activeRecords.contains(mappings.get(lbq))) {
				mappings.remove(lbq);
			}
		}
		this.lbtable = lbtable;
		
	}

	// TODO: implement proper logging. Right now everything goes STDOUT...
	public void log(String toLog) {
		System.out.println(toLog);
	}

	/**
	 * @return the lastModify
	 */
	public long getLastModify() {
		return lastModify;
	}

	/**
	 * @param lastModify
	 *          the lastModify to set
	 */
	public synchronized void setLastModify(long lastModify) {
		this.lastModify = lastModify;
	}

	/**
	 * Entry point
	 * @param args
	 */
	@SuppressWarnings("static-access")
	public static void main(String args[]) {
		String myDbHost = dbHost;
		String myDbUser = dbUser;
		String myDbPwd = dbPwd;
		String myDb = db;
		if (args.length > 0) {
			Options options = new Options();
			Option host = OptionBuilder.withArgName("host").hasArg().withDescription(
					"dhost running the database").create("dbhost");
			Option user = OptionBuilder.withArgName("username").hasArg()
					.withDescription("database user").create("dbuser");
			Option pass = OptionBuilder.withArgName("password").hasArg()
					.withDescription("database password").create("dbpass");
			Option db = OptionBuilder.withArgName("database").hasArg()
					.withDescription("database to use").create("db");
			Option polltime = OptionBuilder.withArgName("milliseconds").hasArg()
					.withDescription("the time interval between database polls").create(
							"polltime");
			Option help = OptionBuilder.withDescription("help").create("help");
			options.addOption(host);
			options.addOption(user);
			options.addOption(pass);
			options.addOption(db);
			options.addOption(polltime);
			options.addOption(help);
			HelpFormatter formatter = new HelpFormatter();
			try {
				CommandLineParser parser = new GnuParser();
				CommandLine cmd = parser.parse(options, args);
				if (cmd.hasOption("dbhost")) {
					myDbHost = cmd.getOptionValue("dbhost");
				}
				if (cmd.hasOption("dbuser")) {
					myDbUser = cmd.getOptionValue("dbuser");
				}
				if (cmd.hasOption("dbpass")) {
					myDbPwd = cmd.getOptionValue("dbpass");
				}
				if (cmd.hasOption("db")) {
					myDb = cmd.getOptionValue("db");
				}
				if (cmd.hasOption("polltime")) {
					checkInterval = Long.parseLong(cmd.getOptionValue("polltime"));
				}
				if (cmd.hasOption("help")) {
					formatter.printHelp("DirectorImpl", options);
					System.exit(1);
				}

			} catch (ParseException pe) {
				System.err.println("Invalid parameters.  Reason: " + pe.getMessage());
				formatter.printHelp("DirectorImpl", options);
				System.exit(1);
			}
		}
		try {
			String name = "Director";
			Director director = new DirectorImpl(new DBTools(myDbHost, myDbUser,
					myDbPwd, myDb));
			// DirectorIf stub =
			// (DirectorIf) UnicastRemoteObject.exportObject(director, 0);
			Registry registry = LocateRegistry.createRegistry(PORT);
			registry.rebind(name, director);
			System.out.println("Director bound");
			System.out.println(registry);

		} catch (Exception e) {
			System.err.println("Director exception:");
			e.printStackTrace();
			System.exit(1);
		}
	}

	/*
	 * (non-Javadoc)
	 * @see com.headcaselabs.lb.common.Director#addGroup(java.lang.String)
	 */
	public void addGroup(String groupName) throws RemoteException {
		if (!servers.containsKey(groupName)) {
			servers.put(groupName, new LinkedList<Server>());
			log("Added group:" + groupName + " for monitoring.");
		}
	}

	/*
	 * (non-Javadoc)
	 * @see com.headcaselabs.lb.common.Director#resolve(com.headcaselabs.lb.common.LBQuery)
	 */
	public Record resolve(LBQuery q) throws RemoteException {
		if (mappings.containsKey(q)) {
			log("Mapping found, reserving"+mappings.get(q).getPdnsString()+" to client:"+q.getSourceIp());
			return mappings.get(q);
		}
		Record record =	lbtable.get(q.getName()).next();
		mappings.put(q, record);
		log("Serving mapping"+mappings.get(q).getPdnsString()+" to client:"+q.getSourceIp());
		return record;
	}
	
	/*
	 * (non-Javadoc)
	 * @see com.headcaselabs.lb.common.Director#addNode(java.lang.String, java.lang.String, java.lang.String, java.lang.String)
	 */
	public void addNode(String groupName,String mode, String arg, String hostname) throws RemoteException {
		log("Received addNode request from client with parameters:"+groupName+" , "+mode+" ,"+arg+" , "+hostname);
		dbtools.insertNode(groupName, mode,arg,System.currentTimeMillis());
		checkForUpdates();
	}
	
	/*
	 * (non-Javadoc)
	 * @see com.headcaselabs.lb.common.Director#getMappingDump()
	 */
	public String getMappingDump()  throws RemoteException {
		StringBuffer sb = new StringBuffer();
		for (LBQuery lbQuery : mappings.keySet()) {
			sb.append(lbQuery.toString());
			sb.append("\t");
			sb.append(mappings.get(lbQuery).toString());
			sb.append("\n");
		}
		return sb.toString();
	}
	/*
	 * (non-Javadoc)
	 * @see com.headcaselabs.lb.common.Director#getServerDump()
	 */
	public String getServerDump() throws RemoteException {
		StringBuffer sb = new StringBuffer();
		for (String str : servers.keySet()) {
			sb.append(str);
			sb.append("\n");
			for (Server srv : servers.get(str)) {
				sb.append(srv.toString());
			}
		}
		return sb.toString();
	}
	/*
	 * (non-Javadoc)
	 * @see com.headcaselabs.lb.common.Director#getDBDump()
	 */
	public String getDBDump() throws RemoteException {
		StringBuffer sb = new StringBuffer();
		Hashtable<String,List<Server>> indb = dbtools.loadAllGroups();
		for (String str : indb.keySet()) {
			sb.append(str);
			sb.append("\n");
			for (Server srv : indb.get(str)) {
				sb.append(srv.toString());
			}
		}
		return sb.toString();
	}

}
